package com.tiantang.kafka.producer;

import java.util.Date;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.tiantang.kafka.utils.DateUtil;

/**
 * 
 * @description:功能描述 带回调函数的生产者
 * @author: liujinkun
 * @date 2018年5月27日
 * @version:1.0
 */
public class MyProducerWithCallBack {
	
	private static Logger logger=LoggerFactory.getLogger(MyProducerWithCallBack.class);
	
	public boolean produce(){
		// 1. 生产者配置
		Properties properties=new Properties();
		// kafka 集群地址配置(注意：这里是Kafka的集群，端口号9092是Kafka的默认端口号，不要写成zookeeper的端口号)
		properties.put("bootstrap.servers", "com.tiantang.hadoop01:9092");
		
		// 发送的消息是否等待应答
		properties.put("acks", "all");
		
		// 重新发送次数
		properties.put("retries", 0);
		
		// 批处理大小
		properties.put("batch.size", 16384);
		
		// 延迟发送时间
		properties.put("linger.ms", 5);
		
		// 内存换冲区
		properties.put("buffer.memory", 33554432);
		
		// key 序列化类
		properties.put("key.serializer", org.apache.kafka.common.serialization.StringSerializer.class);
		
		// 指定分区
		properties.put("partitioner.class", com.tiantang.kafka.partitioner.MyPartitioner.class);
		
		// value 序列化类
		properties.put("value.serializer", org.apache.kafka.common.serialization.StringSerializer.class);
		// 2. 创建KafkaProducer实例
		KafkaProducer<String,String> producer=new KafkaProducer<String,String>(properties);
		
		// 3. 生产消息
		for(int i=0;i<50;i++){
			// 在发送消息时手动指定分区，这时候在properties中配置的分区计算的方式将失效
			producer.send(new ProducerRecord<String, String>("test", 0, "key"+i, i+"-"+DateUtil.dateToStr(new Date(), "yyyy-MM-dd HH:mm:ss")), new Callback() {
				
				public void onCompletion(RecordMetadata metadata, Exception exception) {
					if(metadata != null){
						logger.info("发送消息成功，消息的partitaion={},offset={}",metadata.partition(),metadata.offset());
					}
				}
			});
		}
		// 4. 关闭资源
		producer.close();
		return true;
	}

}
